//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package org.apache.rocketmq.spring.support;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanDefinitionCustomizer;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.util.StringUtils;

public class RocketMQMessageListenerContainerRegistrar implements ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(RocketMQMessageListenerContainerRegistrar.class);
    private ConfigurableApplicationContext applicationContext;
    private final AtomicLong counter = new AtomicLong(0L);
    private final ConfigurableEnvironment environment;
    private final RocketMQProperties rocketMQProperties;
    private final RocketMQMessageConverter rocketMQMessageConverter;

    public RocketMQMessageListenerContainerRegistrar(RocketMQMessageConverter rocketMQMessageConverter, ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties) {
        this.rocketMQMessageConverter = rocketMQMessageConverter;
        this.environment = environment;
        this.rocketMQProperties = rocketMQProperties;
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext)applicationContext;
    }

    public void registerContainer(String beanName, Object bean, RocketMQMessageListener annotation) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " cannot be both instance of " + RocketMQListener.class.getName() + " and " + RocketMQReplyListener.class.getName());
        } else if (!RocketMQListener.class.isAssignableFrom(bean.getClass()) && !RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());
        } else {
            String consumerGroup = this.environment.resolvePlaceholders(annotation.consumerGroup());
            String topic = this.environment.resolvePlaceholders(annotation.topic());
            boolean listenerEnabled = (Boolean)((Map)this.rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)).getOrDefault(topic, true);
            if (!listenerEnabled) {
                log.debug("Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.", consumerGroup, topic);
            } else {
                this.validate(annotation);
                String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(), this.counter.incrementAndGet());
                GenericApplicationContext genericApplicationContext = (GenericApplicationContext)this.applicationContext;
                genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class, () -> {
                    return this.createRocketMQListenerContainer(containerBeanName, bean, annotation);
                }, new BeanDefinitionCustomizer[0]);
                DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer)genericApplicationContext.getBean(containerBeanName, DefaultRocketMQListenerContainer.class);
                if (!container.isRunning()) {
                    try {
                        container.start();
                    } catch (Exception var12) {
                        log.error("Started container failed. {}", container, var12);
                        throw new RuntimeException(var12);
                    }
                }

                log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
            }
        }
    }

    private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean, RocketMQMessageListener annotation) {
        DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
        container.setRocketMQMessageListener(annotation);
        String nameServer = this.environment.resolvePlaceholders(annotation.nameServer());
        nameServer = StringUtils.hasLength(nameServer) ? nameServer : this.rocketMQProperties.getNameServer();
        String accessChannel = this.environment.resolvePlaceholders(annotation.accessChannel());
        container.setNameServer(nameServer);
        if (StringUtils.hasLength(accessChannel)) {
            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        container.setTopic(this.environment.resolvePlaceholders(annotation.topic()));
        String tags = this.environment.resolvePlaceholders(annotation.selectorExpression());
        if (StringUtils.hasLength(tags)) {
            container.setSelectorExpression(tags);
        }

        container.setConsumerGroup(this.environment.resolvePlaceholders(annotation.consumerGroup()));
        container.setTlsEnable(this.environment.resolvePlaceholders(annotation.tlsEnable()));
        if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {
            container.setRocketMQListener((RocketMQListener)bean);
        } else if (RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            container.setRocketMQReplyListener((RocketMQReplyListener)bean);
        }

        container.setMessageConverter(this.rocketMQMessageConverter.getMessageConverter());
        container.setName(name);
        String namespace = this.environment.resolvePlaceholders(annotation.namespace());
        container.setNamespace(RocketMQUtil.getNamespace(namespace, this.rocketMQProperties.getConsumer().getNamespace()));
        return container;
    }

    private void validate(RocketMQMessageListener annotation) {
        if (annotation.consumeMode() == ConsumeMode.ORDERLY && annotation.messageModel() == MessageModel.BROADCASTING) {
            throw new BeanDefinitionValidationException("Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }
}
